package bigdata.hermesfuxi.eagle.etl.utils;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;

import java.util.Arrays;

/**
 * Flink工具类
 * 
 * 该类提供Flink流处理环境的配置和Kafka数据源的创建功能。
 * 主要功能包括：
 * 1. 初始化Flink执行环境
 * 2. 配置检查点和状态后端
 * 3. 创建Kafka数据源（支持两种反序列化方式）
 * 4. 提供统一的Flink作业配置管理
 * 
 * 使用场景：
 * - 实时数据处理作业的启动和配置
 * - Kafka数据源的统一管理
 * - Flink作业的容错和状态管理配置
 * 
 * @author hermesfuxi
 * @version 1.0
 * @since 2021
 */
public class FlinkUtils {
    /**
     * Flink流处理执行环境
     * 全局静态变量，供整个应用使用
     */
    public static StreamExecutionEnvironment env;
    

    /**
     * 静态初始化块
     * 在类加载时初始化Flink执行环境
     */
    static {
        env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setMaxParallelism(128);

    }

    /**
     * 创建Kafka数据源（使用DeserializationSchema）
     *  
     * 该方法用于创建从Kafka消费数据的DataStream，支持自定义反序列化器。
     * 会自动配置检查点、状态后端等Flink作业参数。
     * 
     * @param <T> 数据类型泛型
     * @param args 命令行参数数组，args[0]为Flink配置文件路径，args[1]为Kafka配置文件路径
     * @param deserializer 反序列化器类
     * @return 配置好的DataStream数据流
     * @throws Exception 配置或创建过程中的异常
     */
    public static <T> DataStream<T> getKafkaSource(String[] args, Class<? extends DeserializationSchema<T>> deserializer) throws Exception {
        // 从配置文件中读取Flink相关参数
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        // 获取检查点间隔时间，默认10秒
        long checkpointInterval = parameterTool.getLong("flink.checkpoint.interval", 10000L);
        // 获取状态后端存储路径
        String stateBackEndPath = parameterTool.get("flink.stateBackEnd.path");

        // 启用检查点，使用精确一次语义
        env.enableCheckpointing(checkpointInterval, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 设置检查点外部化，程序被cancel后保留Checkpoint数据，可根据需要恢复到指定Checkpoint
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置RocksDB作为状态后端，支持大状态存储
        env.setStateBackend(new RocksDBStateBackend(stateBackEndPath));

        // 使用反射创建反序列化器实例
        DeserializationSchema<T> deserializationSchema = deserializer.newInstance();
        // JDK9之后推荐使用的方式（当前方式在JDK9+中已过时）
//        DeserializationSchema<T> deserializationSchema = deserializer.getDeclaredConstructor().newInstance();

        // 从Kafka配置文件中读取参数
        ParameterTool kafkaParameterTool = ParameterTool.fromPropertiesFile(args[1]);
        // 获取要消费的Topic列表
        String[] topics = kafkaParameterTool.get("topics").split(",");
        // 创建Kafka消费者
        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<T>(Arrays.asList(topics), deserializationSchema, kafkaParameterTool.getProperties());
        // 设置不将偏移量提交到Kafka的特殊topic中，由Flink自己管理偏移量
        kafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        return env.addSource(kafkaConsumer);
    }

    /**
     * 创建Kafka数据源（使用KafkaDeserializationSchema）
     * 
     * 该方法用于创建从Kafka消费数据的DataStream，支持读取Kafka消息的元数据信息
     * （如topic、partition、offset等）。相比getKafkaSource方法，此方法可以获取
     * 更丰富的Kafka消息上下文信息。
     * 
     * 主要特点：
     * 1. 支持获取Kafka消息的topic、partition、offset等元数据
     * 2. 适用于需要根据消息来源进行不同处理的场景
     * 3. 提供更灵活的Kafka消息反序列化能力
     * 
     * @param <T> 数据类型泛型
     * @param args 命令行参数数组，args[0]为Flink配置文件路径，args[1]为Kafka配置文件路径
     * @param deserializer Kafka反序列化器类
     * @return 配置好的DataStream数据流
     * @throws Exception 配置或创建过程中的异常
     */
    public static <T> DataStream<T> getKafkaSourceV2(String[] args, Class<? extends KafkaDeserializationSchema<T>> deserializer) throws Exception {
        // 从配置文件中读取Flink相关参数
        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        // 获取检查点间隔时间，默认30秒
        long checkpointInterval = parameterTool.getLong("flink.checkpoint.interval", 30000L);
        // 获取状态后端存储路径
        String stateBackEndPath = parameterTool.get("flink.stateBackEnd.path");

        // 启用检查点，使用精确一次语义
        env.enableCheckpointing(checkpointInterval, CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 设置检查点外部化，程序被cancel后保留Checkpoint数据，可根据需要恢复到指定Checkpoint
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置RocksDB作为状态后端，支持大状态存储
        env.setStateBackend(new RocksDBStateBackend(stateBackEndPath));

        // 使用反射创建Kafka反序列化器实例
        KafkaDeserializationSchema<T> deserializationSchema = deserializer.newInstance();

        // 从Kafka配置文件中读取参数
        ParameterTool kafkaParameterTool = ParameterTool.fromPropertiesFile(args[1]);
        // 获取要消费的Topic列表
        String[] topics = kafkaParameterTool.get("topics").split(",");
        // 创建Kafka消费者，使用KafkaDeserializationSchema
        FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<T>(Arrays.asList(topics), deserializationSchema, kafkaParameterTool.getProperties());

        // 设置不将偏移量提交到Kafka的特殊topic中，由Flink自己管理偏移量
        kafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        return env.addSource(kafkaConsumer);
    }
}

